﻿using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using org.apache.zookeeper;
using org.apache.zookeeper.data;
using static org.apache.zookeeper.Watcher.Event;
using static org.apache.zookeeper.ZooKeeper;

namespace ESoft.Zookeeper
{
    public class ZookeeperClient
    {
        public ZooKeeper ZK { get; set; }

        // 配置项
        public string QueryPath { get; set; } = "/Configuration";

        //节点状态信息
        public Stat Stat { get; set; }

        // 配置数据
        public byte[] ConfigData { get; set; } = null;
        private string serviceAddress;
        private int timeout;
        static Semaphore sema = new Semaphore(1, 1);
        public ZookeeperClient(string serviceAddress, int timeout)
        {
            this.serviceAddress = serviceAddress;
            this.timeout = timeout;
            Conn();
        }

        public async Task<string> CreateAsync(string path, byte[] data, List<ACL> acl, CreateMode createMode)
        {
            Conn();
            return await ZK.createAsync(path, data, acl, createMode);
        }

        public async Task DeleteAsync(string path, int version = -1)
        {
            Conn();
            await ZK.deleteAsync(path, version);
        }

        public async Task<Stat> ExistsAsync(string path)
        {
            Conn();
            return await ZK.existsAsync(path);
        }

        public async Task<ChildrenResult> GetChildrenAsync(string path)
        {
            Conn();
            return await ZK.getChildrenAsync(path);
        }

        public async Task<DataResult> GetDataAsync(string path)
        {
            Conn();
            return await ZK.getDataAsync(path);
        }

        // 读取节点的配置数据
        public async Task<string> ReadConfigDataAsync()
        {
            if (this.ZK == null)
            {
                return string.Empty;
            }

            var stat = await ZK.existsAsync(QueryPath, true);

            if (stat == null)
            {
                return string.Empty;
            }

            this.Stat = stat;

            var dataResult = await ZK.getDataAsync(QueryPath, true);

            return Encoding.UTF8.GetString(dataResult.Data);
        }

        private void Conn()
        {
            if (ZK == null || ZK.getState() == States.CLOSED
            || ZK.getState() == States.NOT_CONNECTED)
            {
                sema.WaitOne();
                ZK = new ZooKeeper(serviceAddress, timeout, new ConfigServiceWatcher(this));
            }
        }


        public class ConfigServiceWatcher : Watcher
        {
            private ZookeeperClient _cs = null;

            public ConfigServiceWatcher(ZookeeperClient cs)
            {
                _cs = cs;
            }

            public override Task process(WatchedEvent @event)
            {
                if (@event.getState() == KeeperState.SyncConnected)
                {
                    sema.Release();
                }
                return Task.CompletedTask;
                //if (@event.get_Type() == EventType.NodeDataChanged)
                //{
                //    var data = await _cs.ReadConfigDataAsync();
                //}
            }
        }


    }
}
